package com.cloud.rocketmq;

import com.cloud.configure.RocketMqConfig;
import com.cloud.exception.CommonException;
import com.cloud.exception.CustomException;
import com.cloud.message.DelayLevel;
import com.cloud.message.MQMessageBuilder;
import com.cloud.utils.StringUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Objects;

/**
 * RocketMQ 生产者对象
 *
 * @author yzj
 */
@Component
public class RocketMqProducer {

    @Autowired
    private RocketMqConfig rocketMqConfig;

    /**
     * 同步方式发送消息。此方法仅在发送过程完全完成时返回。
     * 处理程序 MQClientException、RemotingException、MQBrokerException
     */
    public SendResult sendSync(String topic, String tag, Object message, String key) {
        try {
            DefaultMQProducer producer = rocketMqConfig.mqProducer();
            if (StringUtils.isEmpty(key)) {
                key = System.currentTimeMillis() + "";
            }
            return producer.send(MQMessageBuilder.message(topic, tag, message, key).build());
        } catch (Exception e) {
            throw new CustomException(CommonException.ROCKET_MQ_SEND_ERROR.getCode(), "同步发送消息异常:" + e);
        }
    }

    /**
     * 以同步方式发送消息。此方法仅在发送过程完全完成时返回。支持延迟发送
     * 处理程序 MQClientException、RemotingException、MQBrokerException
     */
    public SendResult sendSync(String topic, String tag, Object message, String key, DelayLevel level) {
        try {
            DefaultMQProducer producer = rocketMqConfig.mqProducer();
            if (StringUtils.isEmpty(key)) {
                key = System.currentTimeMillis() + "";
            }
            return producer.send(MQMessageBuilder.message(topic, tag, message, key, level).build());
        } catch (Exception e) {
            throw new CustomException(CommonException.ROCKET_MQ_SEND_ERROR.getCode(), "同步发送消息异常:" + e);
        }
    }

    /**
     * 指定消息队列发送同步消息。
     * 处理程序 MQClientException、RemotingException、MQBrokerException
     */
    public SendResult sendSyncOrderByKey(String topic, String tag, Object message, String key, Object orderBy) {
        try {
            DefaultMQProducer producer = rocketMqConfig.mqProducer();
            if (StringUtils.isEmpty(key)) {
                key = System.currentTimeMillis() + "";
            }
            Objects.requireNonNull(orderBy);
            return producer.send(MQMessageBuilder.message(topic, tag, message, key).build(),
                    new SelectMessageQueueByHash(), orderBy);
        } catch (Exception e) {
            throw new CustomException(CommonException.ROCKET_MQ_SEND_ERROR.getCode(), "指定消息队列发送消息异常:" + e);
        }
    }

    /**
     * 指定消息队列发送同步消息。支持延迟发送
     * 处理程序 MQClientException、RemotingException、MQBrokerException
     */
    public SendResult sendSyncOrderByKey(String topic, String tag, Object message, String key, Object orderBy, DelayLevel level) {
        try {
            DefaultMQProducer producer = rocketMqConfig.mqProducer();
            if (StringUtils.isEmpty(key)) {
                key = System.currentTimeMillis() + "";
            }
            Objects.requireNonNull(orderBy);
            return producer.send(MQMessageBuilder.message(topic, tag, message, key, level).build(),
                    new SelectMessageQueueByHash(), orderBy);
        } catch (Exception e) {
            throw new CustomException(CommonException.ROCKET_MQ_SEND_ERROR.getCode(), "指定消息队列发送消息异常:" + e);
        }
    }

    /**
     * 异步发送消息。此方法立即返回
     * 处理程序 MQClientException、RemotingException、MQBrokerException
     */
    public void sendAsync(String topic, String tag, Object message, String key) {
        try {
            DefaultMQProducer producer = rocketMqConfig.mqProducer();
            if (StringUtils.isEmpty(key)) {
                key = System.currentTimeMillis() + "";
            }
            Objects.requireNonNull(key);
            producer.send(MQMessageBuilder.message(topic, tag, message, key).build());
        } catch (Exception e) {
            throw new CustomException(CommonException.ROCKET_MQ_SEND_ERROR.getCode(), "异步消息发送异常:" + e);
        }
    }

    /**
     * 异步发送消息。此方法立即返回。发送完成后，将执行SendCallback。
     * 处理程序 MQClientException、RemotingException、MQBrokerException
     */
    public void sendAsync(String topic, String tag, Object message, String key, SendCallback callback) {
        try {
            DefaultMQProducer producer = rocketMqConfig.mqProducer();
            if (StringUtils.isEmpty(key)) {
                key = System.currentTimeMillis() + "";
            }
            Objects.requireNonNull(key);
            producer.send(MQMessageBuilder.message(topic, tag, message, key).build(), callback);
        } catch (Exception e) {
            throw new CustomException(CommonException.ROCKET_MQ_SEND_ERROR.getCode(), "异步消息发送异常:" + e);
        }
    }

    /**
     * 异步发送消息。此方法立即返回。发送完成后，将执行SendCallback。支持延迟发送
     * 处理程序 MQClientException、RemotingException、MQBrokerException
     */
    public void sendAsync(String topic, String tag, Object message, String key, SendCallback callback, DelayLevel level) {
        try {
            DefaultMQProducer producer = rocketMqConfig.mqProducer();
            if (StringUtils.isEmpty(key)) {
                key = System.currentTimeMillis() + "";
            }
            producer.send(MQMessageBuilder.message(topic, tag, message, key, level).build(), callback);
        } catch (Exception e) {
            throw new CustomException(CommonException.ROCKET_MQ_SEND_ERROR.getCode(), "异步消息发送异常:" + e);
        }
    }

    /**
     * 指定消息队列发送异步消息。发送完成后，将执行SendCallback。
     * 处理程序 MQClientException、RemotingException、MQBrokerException
     */
    public void sendAsyncOrderByKey(String topic, String tag, Object message, String key, Object orderBy, SendCallback callback) {
        try {
            DefaultMQProducer producer = rocketMqConfig.mqProducer();
            if (StringUtils.isEmpty(key)) {
                key = System.currentTimeMillis() + "";
            }
            Objects.requireNonNull(orderBy);
            producer.send(MQMessageBuilder.message(topic, tag, message, key).build(), new SelectMessageQueueByHash(), orderBy, callback);
        } catch (Exception e) {
            throw new CustomException(CommonException.ROCKET_MQ_SEND_ERROR.getCode(), "指定消息队列发送异步消息异常:" + e);
        }
    }

    /**
     * 指定消息队列发送异步消息。发送完成后，将执行SendCallback。支持延迟发送
     * 处理程序 MQClientException、RemotingException、MQBrokerException
     */
    public void sendAsyncOrderByKey(String topic, String tag, Object message, String key, Object orderBy,
                                    SendCallback callback, DelayLevel level) {
        try {
            DefaultMQProducer producer = rocketMqConfig.mqProducer();
            if (StringUtils.isEmpty(key)) {
                key = System.currentTimeMillis() + "";
            }
            Objects.requireNonNull(orderBy);
            producer.send(MQMessageBuilder.message(topic, tag, message, key, level).build(),
                    new SelectMessageQueueByHash(), orderBy, callback);
        } catch (Exception e) {
            throw new CustomException(CommonException.ROCKET_MQ_SEND_ERROR.getCode(), "指定消息队列发送异步消息异常:" + e);
        }
    }

    /**
     * 单向消息同步发送。它具有最大的吞吐量，但也有可能丢失消息。
     * 处理程序 MQClientException、RemotingException、MQBrokerException
     */
    public void sendOneSync(String topic, String tag, Object message, String key) {
        try {
            DefaultMQProducer producer = rocketMqConfig.mqProducer();
            if (StringUtils.isEmpty(key)) {
                key = System.currentTimeMillis() + "";
            }
            producer.sendOneway(MQMessageBuilder.message(topic, tag, message, key).build());
        } catch (Exception e) {
            throw new CustomException(CommonException.ROCKET_MQ_SEND_ERROR.getCode(), "单向消息发送异常:" + e);
        }
    }

    /**
     * 单向消息同步发送。支持延迟发送
     * 处理程序 MQClientException、RemotingException、MQBrokerException
     */
    public void sendOneSync(String topic, String tag, Object message, String key, DelayLevel level) {
        try {
            DefaultMQProducer producer = rocketMqConfig.mqProducer();
            if (StringUtils.isEmpty(key)) {
                key = System.currentTimeMillis() + "";
            }
            producer.sendOneway(MQMessageBuilder.message(topic, tag, message, key, level).build());
        } catch (Exception e) {
            throw new CustomException(CommonException.ROCKET_MQ_SEND_ERROR.getCode(), "单向消息发送异常:" + e);
        }
    }

    /**
     * 单向消息异步发送。
     * 处理程序 MQClientException、RemotingException、MQBrokerException
     */
    public void sendOneAsyncOrderByKey(String topic, String tag, Object message, String key, Object orderBy) {
        try {
            DefaultMQProducer producer = rocketMqConfig.mqProducer();
            if (StringUtils.isEmpty(key)) {
                key = System.currentTimeMillis() + "";
            }
            Objects.requireNonNull(orderBy);
            producer.sendOneway(MQMessageBuilder.message(topic, tag, message, key).build(), new SelectMessageQueueByHash(), orderBy);
        } catch (Exception e) {
            throw new CustomException(CommonException.ROCKET_MQ_SEND_ERROR.getCode(), "单向消息异步发送异常:" + e);
        }
    }

    /**
     * 单向消息异步发送。支持延迟发送
     * 处理程序 MQClientException、RemotingException、MQBrokerException
     */
    public void sendOneAsyncOrderByKey(String topic, String tag, Object message, String key, Object orderBy, DelayLevel level) {
        try {
            DefaultMQProducer producer = rocketMqConfig.mqProducer();
            if (StringUtils.isEmpty(key)) {
                key = System.currentTimeMillis() + "";
            }
            Objects.requireNonNull(orderBy);
            producer.sendOneway(MQMessageBuilder.message(topic, tag, message, key, level).build(),
                    new SelectMessageQueueByHash(), orderBy);
        } catch (Exception e) {
            throw new CustomException(CommonException.ROCKET_MQ_SEND_ERROR.getCode(), "单向消息异步发送异常:" + e);
        }
    }

    /**
     * 事务消息发送。不支持延迟发送和批量发送
     */
    public void sendMessageTransaction(String topic, String tag, Object message, String key) {
        try {
            if (StringUtils.isEmpty(key)) {
                key = System.currentTimeMillis() + "";
            }
            TransactionMQProducer producer = rocketMqConfig.getProducer(topic + "_" + tag);
            producer.sendMessageInTransaction(MQMessageBuilder.message(topic, tag, message, key).build(), message);
        } catch (Exception e) {
            throw new CustomException(CommonException.ROCKET_MQ_SEND_ERROR.getCode(), "事务消息发送异常:" + e);
        }
    }

}
